package com.atguigu.flink.chapter05.Transform;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/8/9 10:30
 */
public class Flink03_FlatMap {
    public static void main(String[] args) throws Exception {
        // 0.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
                .fromElements(
                        Arrays.asList(
                                new WaterSensor("sensor_1", 123L, 1),
                                new WaterSensor("sensor_2", 1234L, 11),
                                new WaterSensor("sensor_3", 1235L, 21)

                        )
                )
                .flatMap(new FlatMapFunction<List<WaterSensor>, WaterSensor>() {
                    @Override
                    public void flatMap(List<WaterSensor> value, Collector<WaterSensor> out) throws Exception {
                        for (WaterSensor waterSensor : value) {
                            if (waterSensor.getVc() > 30) {
                                out.collect(waterSensor);
                            }
                        }
                    }
                })
                .print();


        env.execute();
    }
}
/*
    flatmap: 一进多出、或 零出 ，取决于调用了几次 colletor

    提供了富函数版本： RichFlatMapFucntion
 */